package comp6231.project.frontend;

import java.math.BigInteger;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.Iterator;

import comp6231.project.common.Constants;
import comp6231.project.common.Message;
import comp6231.project.common.ReplicaState;
import comp6231.project.common.Utils;

public class ReplicaManager extends Thread {

	class TableRow {
		public Message msg;
		
		// results - a map from replicaID to its result
		public Hashtable<String, Integer> results = new Hashtable<String, Integer>();
		
		public Integer finalResult = null;
	}
	
	private Hashtable<BigInteger, TableRow> table = new Hashtable<BigInteger, TableRow>();
	private ArrayList<Message> messageQueue = new ArrayList<Message>();
	
	// keep track of Replica errors - a map from replicaID to error count
	private Hashtable<String, Integer> errorsTracking = new Hashtable<String, Integer>();
	
	private Hashtable<BigInteger, Object> resultLocks = new Hashtable<BigInteger, Object>();
	
	private String groupName = "";
	private int multicastPort = 10100;
	private int rmPort;
	
	private boolean onHold = false;
	
	private String hostDRS1 = "";
	private String hostDRS2 = "";
	private String hostDRS3 = "";
	
	/**
	 * Constructor
	 * @param hostDRS1 IP address of DRS1
	 * @param hostDRS2 IP address of DRS2
	 * @param hostDRS3 IP address of DRS3
	 */
	public ReplicaManager(int rmPort, String groupName, String hostDRS1, String hostDRS2, String hostDRS3) {
		this.hostDRS1 = hostDRS1;
		this.hostDRS2 = hostDRS2;
		this.hostDRS3 = hostDRS3;
		this.rmPort = rmPort;
		this.groupName = groupName;
	}
	
	private synchronized void setOnHold(boolean onHold) {
		this.onHold = onHold;
	}
	
	private synchronized boolean isOnHold() {
		return onHold;
	}
	
	private synchronized void addToTable(Message msg) {
		TableRow tr = new TableRow();
		tr.msg = msg;
		table.put(msg.sequenceNum, tr);
	}

	private TableRow getTableRow(BigInteger seqNum) {
		synchronized (table) {
			if (table.containsKey(seqNum)) {
				return table.get(seqNum);
			}
		}
		System.out.println("Error: No table row for seqNum " + seqNum.toString());
		return null;
	}
	
	private void removeTableRow(BigInteger seqNum) {
		synchronized (table) {
			if (table.containsKey(seqNum)) {
				table.remove(seqNum);
			}
		}		
	}
	
	private synchronized void addToQueue(Message msg) {
		messageQueue.add(msg);
	}
	
	private synchronized void processQueuedMessage() {
		for (int i = 0; i < messageQueue.size(); i++) {
			multicastMessageToReplica(messageQueue.get(i), false);
		}
		messageQueue.clear();
	}

	/**
	 * Multi-cast to all replica in the group
	 * @param msg Message to send
	 * @param addTable false if it is a re-send
	 */
	private void multicastMessageToReplica(Message msg, boolean addTable) {
		DatagramSocket dsOut = null;
		try {
			
			// Simulate message lost
			if (addTable && 
					msg.request.equals(Message.REQUEST_BUY) &&
					msg.itemID.equals("1115") &&
					msg.sequenceNum.intValue() == 1) {
				System.out.println("Simulate message lost seq# " + msg.sequenceNum + " BUY itemID=1115");
				
			} else {
				
			InetAddress group = InetAddress.getByName(groupName);
			byte bufOut[] = Utils.toBytes(msg);
			DatagramPacket packOut = new DatagramPacket(
					bufOut, bufOut.length, group, Constants.RM_MULTICAST_PORT);
			dsOut = new DatagramSocket();
			dsOut.send(packOut);
			
			System.out.println("Multicast msg " + msg.request + " seq# " + msg.sequenceNum + " to group " + groupName);
			
			}
			
			if (addTable) {
				addToTable(msg);
			}
			
		} catch (Exception ex) {
			ex.printStackTrace();

		} finally {
			if (dsOut != null) {
				dsOut.close();
			}
		}				
	}

	private void askDRSToReplaceReplica(String replicaID) {
		DatagramSocket ds = null;
		try {
			String drsHostReplace;
			String drsHostGetState;
			String replicaIDGetState;
			if (replicaID.endsWith("1")) {
				drsHostReplace = hostDRS1;
				drsHostGetState = hostDRS2;
				replicaIDGetState = replicaID.replace("1", "2");
			} else if (replicaID.endsWith("2")) {
				drsHostReplace = hostDRS2;
				drsHostGetState = hostDRS1;
				replicaIDGetState = replicaID.replace("2", "1");
			} else {
				drsHostReplace = hostDRS3;	
				drsHostGetState = hostDRS1;
				replicaIDGetState = replicaID.replace("3", "1");
			}
			
			System.out.println("Ask DRS host " + drsHostGetState + " for state of replica " + replicaIDGetState);
			
			Message msg = new Message();
			msg.request = Message.REQUEST_GETREPLSTATE;
			msg.replicaID = replicaIDGetState;
			
			InetAddress hostGetState = InetAddress.getByName(drsHostGetState);
			byte bufOut[] = Utils.toBytes(msg);
			DatagramPacket packOut = new DatagramPacket(
					bufOut, bufOut.length, hostGetState, Constants.DRS_PORT);
			ds = new DatagramSocket();
			ds.send(packOut);
			
			byte bufIn[] = new byte[Utils.UDP_DATAGRAM_MAX_SIZE];
			DatagramPacket pack = new DatagramPacket(bufIn,
					bufIn.length);
			ds.receive(pack);
			
			ReplicaState state = (ReplicaState)Utils.fromBytes(bufIn);
			System.out.println("Received state from " + drsHostGetState + ": " + state.toString());
			
			System.out.println("Ask DRS host " + drsHostReplace + " to replace replica " + replicaID);
			
			msg.request = Message.REQUEST_REPLREPLICA;
			msg.replicaID = replicaID;
			bufOut = Utils.toBytes(msg);
			
			InetAddress hostReplace = InetAddress.getByName(drsHostReplace);
			packOut = new DatagramPacket(
					bufOut, bufOut.length, hostReplace, Constants.DRS_PORT);
			ds.send(packOut);
			
			bufOut = Utils.toBytes(state);
			packOut = new DatagramPacket(
					bufOut, bufOut.length, hostReplace, Constants.DRS_PORT);
			ds.send(packOut);
			
			// Wait for notification from DRS
			bufIn = new byte[Utils.UDP_DATAGRAM_MAX_SIZE];
			pack = new DatagramPacket(bufIn,
					bufIn.length);
			ds.receive(pack);
			Object obj = Utils.fromBytes(bufIn);
			if (obj instanceof Message) {
				msg = (Message)obj;
				if (!msg.request.equals(Message.REQUEST_REPL_ACK)) {
					System.out.println("Error: Replica replacemenet failed " + msg.replicaID);
				}
			}
			
		} catch (Exception ex) {
			ex.printStackTrace();

		} finally {
			if (ds != null) {
				ds.close();
			}
		}				
	}
	
	private void increaseErrorCount(String replicaID) {
		if (!errorsTracking.containsKey(replicaID)) {
			errorsTracking.put(replicaID, new Integer(0));
		}
		Integer i = errorsTracking.get(replicaID);
		Integer newCount = new Integer(i.intValue() + 1);
		errorsTracking.put(replicaID, newCount);
		System.out.println("Error count for replica " + replicaID + " now at " + newCount);
	}
	
	private int getErrorCount(String replicaID) {
		if (errorsTracking.containsKey(replicaID)) { 
			return errorsTracking.get(replicaID).intValue();
		}
		return 0;
	}
	
	private void resetErrorCount(String replicaID) {
		if (errorsTracking.containsKey(replicaID)) { 
			errorsTracking.put(replicaID, new Integer(0));
		}		
	}
	
	private void waitForResultToBeReady(Message msg) {
		Object lock = new Object();
		resultLocks.put(msg.sequenceNum, lock);
		try {
			synchronized(lock) {
				lock.wait();
			}
		} catch (InterruptedException ie) {
			ie.printStackTrace();
		}		
	}
	
	private void signalResultIsReady(Message msg) {
		Object lock = resultLocks.get(msg.sequenceNum);
		synchronized(lock) {
			lock.notify();
		}		
	}
	
	public int processMessageFromSequencer(Message msg) {
		if (isOnHold()) {
			addToQueue(msg);			
		} else {
			processQueuedMessage();
			multicastMessageToReplica(msg, true);
		}
		
		waitForResultToBeReady(msg);		
				
		TableRow tr = getTableRow(msg.sequenceNum);
		Integer r = tr.finalResult;
		removeTableRow(msg.sequenceNum);
		return r.intValue();
	}
	
	private void processMessageFromReplica(Message m) {
		try {
			final Message msg = m;
			
			if (msg.request.equals(Message.REQUEST_NEGACK)) {
				// Spawn a new thread to handle re-send
				Runnable r = new Runnable() {
					public void run() {
						System.out.println("Received NEGACK for seq# " + msg.sequenceNum + " from " + msg.replicaID);
						TableRow tr = getTableRow(msg.sequenceNum);
						multicastMessageToReplica(tr.msg, false);
					}
				};
				(new Thread(r)).start();
				
			} else { 
				// Reply from Replica's operation
			
				// Put received results into proper table row
				TableRow tr = getTableRow(msg.sequenceNum);							
				tr.results.put(msg.replicaID, new Integer(msg.result));
				System.out.println("Received Replica result " + msg.result + " from replicaID " + msg.replicaID + " for message seq# " + msg.sequenceNum);
											
				// If all results for a row have been collected, perform error detection and return result back to FE
				if (tr.results.values().size() == 3) {
					System.out.println("Collected all 3 results from replicas for MSG seq# " + msg.sequenceNum);
					int finalResult = 0;
					String[] replicaIDs = new String[3];
					int[] results = new int[3];
					Iterator<String> it = tr.results.keySet().iterator();
					int i = 0;
					while (it.hasNext()) {
						replicaIDs[i] = it.next();
						results[i] = tr.results.get(replicaIDs[i]).intValue();
						i++;
					}
					
					if (results[0] == results[1] && results[0] == results[2]) {
						finalResult = results[0];
						System.out.println("All 3 results are the same!");
						
					} else if (results[0] != results[1] && results[0] != results[2]) {
						// replicaIDs[0] had error
						System.out.println("Result from " + replicaIDs[0] + " had problem!");
						finalResult = results[1];
						increaseErrorCount(replicaIDs[0]);
						
					} else if (results[1] != results[0] && results[1] != results[2]) {
						// replicaIDs[1] had error
						System.out.println("Result from " + replicaIDs[1] + " had problem!");
						finalResult = results[0];
						increaseErrorCount(replicaIDs[1]);
						
					} else if (results[2] != results[0] && results[2] != results[1]) {
						// replicaIDs[2] had error
						System.out.println("Result from " + replicaIDs[2] + " had problem!");
						finalResult = results[0];
						increaseErrorCount(replicaIDs[2]);
						
					}
					tr.finalResult = new Integer(finalResult);
					
					signalResultIsReady(msg);
					
					for (int j = 0; j < replicaIDs.length; j++) {
						if (getErrorCount(replicaIDs[j]) == 3) {
							setOnHold(true);
							// If 3 errors detected for a Replica, spawn a new thread to handle Replica replacement
							final String replicaIDReplace = replicaIDs[j];
							//Runnable r = new Runnable() {
							//	public void run() {
									System.out.println("Replace replica " + replicaIDReplace);
									askDRSToReplaceReplica(replicaIDReplace);
							//	}
							//};			
							setOnHold(false);
						}
					}
				}							
			}		
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}
	
	public void run() {
		DatagramSocket dsIn = null;
		DatagramSocket dsOut = null;
		try {
			// Wait for messages from Replicas within the group
			dsIn = new DatagramSocket(rmPort);
			while (true) {
				try {
					final byte bufIn[] = new byte[Utils.UDP_DATAGRAM_MAX_SIZE];
					final DatagramPacket pack = new DatagramPacket(bufIn,
							bufIn.length);
					dsIn.receive(pack);
					
					
					Object obj = Utils.fromBytes(bufIn);
					if (obj instanceof Message) {
						System.out.println("Received msg from Replica " + ((Message)obj).result);
						processMessageFromReplica((Message)obj);
					} else {
						System.out.println("Unknown msg!");
					}

				} catch (Exception ex) {
					ex.printStackTrace();
				}
			}
		} catch (Exception ex) {
			ex.printStackTrace();

		} finally {
			if (dsIn != null) {
				dsIn.close();
			}
			if (dsOut != null) {
				dsOut.close();
			}
		}
	}
	
	public static void main(String[] args) {
		/*ReplicaManager rm = new ReplicaManager("localhost", "localhost", "localhost");		
		rm.start();
		try {
			Message m = new Message();
			m.sequenceNum = BigInteger.valueOf(1);
			m.request = Message.REQUEST_BUY;
			m.customerID = "000001";
			m.itemID = "0002";
			m.numItems = 10;			
			int r = rm.processMessageFromSequencer(m);
			System.out.println("Result = " + r);
			rm.join();
		} catch (Exception ex) {
			ex.printStackTrace();
		}*/
	}
}
